package org.com.gamemall.Service.Impl;

import cn.hutool.core.bean.BeanUtil;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.com.gamemall.Entity.UserVoucher;
import org.com.gamemall.Entity.Voucher;
import org.com.gamemall.Mapper.VoucherMapper;
import org.com.gamemall.Service.IUserVoucherService;
import org.com.gamemall.Service.IVoucherService;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Lazy;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.http.ResponseEntity;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

@Slf4j
@Service
public class VoucherServiceImpl extends ServiceImpl<VoucherMapper, Voucher> implements IVoucherService , DisposableBean {

    private static final String LOGIN_USER_KEY = "login:user:";
    private static final String STOCK_KEY = "seckill:stock:";
    private static final String ORDER_KEY = "seckill:order:";
    private static final String USER_ORDER_LOCK = "seckill:user_order_lock:";

    @Autowired
    private StringRedisTemplate redisTemplate;
    @Autowired
    private IUserVoucherService userVoucherService;
    @Autowired
    @Lazy
    private IVoucherService voucherService;
    @Autowired
    private RedissonClient redissonClient;
    @Autowired
    @Qualifier("voucherTaskExecutor")
    private ThreadPoolTaskExecutor voucherTaskExecutor;
    @Autowired
    private RabbitTemplate rabbitTemplate;

    private volatile boolean running = true;

    //      创建lua脚本
    private static final DefaultRedisScript<Long> SECKILL_SCRIPT;

    static {
        SECKILL_SCRIPT = new DefaultRedisScript<>();
        SECKILL_SCRIPT.setLocation(new ClassPathResource("testlua.lua"));
        SECKILL_SCRIPT.setResultType(Long.class);
    }


    @PostConstruct
    public void init() {
voucherTaskExecutor.execute(new voucherInfo());
    }
  @PreDestroy
    @Override
    public void destroy() throws Exception {
        running=false;
        if (voucherTaskExecutor instanceof ExecutorService){
            ((ExecutorService) voucherTaskExecutor).shutdown();
        }
    }
    /**
     监控方法

     */
    @Scheduled(fixedRate = 1000)
    public void monitorThreadPool() {
        log.info("活跃线程数: {}", voucherTaskExecutor.getActiveCount());
        log.info("队列大小: {}", voucherTaskExecutor.getThreadPoolExecutor().getQueue().size());
        log.info("已完成任务数: {}", voucherTaskExecutor.getThreadPoolExecutor().getCompletedTaskCount());
    }

    /**
     * 检查redis连接配置
     */
    private boolean isRedisAvailable() {
        try {
            redisTemplate.hasKey("health_check");
            return true;
        } catch (IllegalStateException e) {
            if (e.getMessage().contains("STOPPED")) {
                log.warn("Redis连接工厂已停止");
                return false;
            }
            throw e;
        } catch (Exception e) {
            log.warn("Redis连接检查失败: {}", e.getMessage());
            return false;
        }
    }

    /**
     * 处理消息队列实现runnable
     */
    private class voucherInfo implements Runnable {
        @Override
        public void run() {
            while (running && !Thread.currentThread().isInterrupted()) {
//读取消息队列中的订单信息
                try {
                    if (!isRedisAvailable()){
                        log.warn("Redis连接失败，请检查Redis服务是否启动");
                        Thread.sleep(3000);
                        continue;
                    }
                    List<MapRecord<String, Object, Object>> read = redisTemplate.opsForStream()
                            .read(Consumer.from("seckill-order-group", "c1"),
                                    StreamReadOptions.empty().count(1).block(Duration.ofSeconds(20)),
                                    StreamOffset.create("stream.order", ReadOffset.lastConsumed()));

                    if (read == null || read.isEmpty()) {
                        log.info("暂无消息");
                        continue;
                    }
                    //解析消息
                    MapRecord<String, Object, Object> entries = read.get(0);
                    Map<Object, Object> value = entries.getValue();
                    UserVoucher orderVoucher = BeanUtil.fillBeanWithMap(value, new UserVoucher(), true);
                    // 创建订单
                    voucherService.handlerOrderInfo(orderVoucher);
                    //ACK消息确认
                    redisTemplate.opsForStream().acknowledge("stream.order", "seckill-order-group", entries.getId());
                }catch (InterruptedException e){
                    Thread.currentThread().interrupt();
                    log.warn("消息处理线程被中断");
                    break;
                } catch (org.springframework.data.redis.RedisSystemException e) {
                    if (e.getMessage().contains("Connection closed") ||
                            e.getCause() instanceof io.lettuce.core.RedisException) {
                        log.warn("Redis连接已关闭");
                          break;
                    }
                    log.error("处理消息异常", e);
                } catch (Exception e){
                    log.error("处理消息异常", e);
                    handleerrorInfo();
                }
            }
            log.info("消息处理完毕");
        }
    }

    /**
     * 处理消息队列错误
     */
    private void handleerrorInfo() {
        int retryCount = 0;
        final int MAX_RETRY = 3;

        while (running && retryCount < MAX_RETRY && !Thread.currentThread().isInterrupted()) {
            try {
                if (!isRedisAvailable()) {
                    log.warn("Redis连接不可用，无法处理错误消息");
                    break;
                }

                List<MapRecord<String, Object, Object>> handle = redisTemplate.opsForStream()
                        .read(Consumer.from("seckill-order-group", "c1"),
                                StreamReadOptions.empty().count(1),
                                StreamOffset.create("stream.order", ReadOffset.from("0")));

                if (handle == null || handle.isEmpty()) {
                    log.info("没有待处理的错误消息");
                    break;
                }

                MapRecord<String, Object, Object> entries = handle.get(0);
                Map<Object, Object> value = entries.getValue();
                UserVoucher userVoucher = BeanUtil.fillBeanWithMap(value, new UserVoucher(), true);
                voucherService.handlerOrderInfo(userVoucher);
                //ACK消息确认
                redisTemplate.opsForStream().acknowledge("stream.order", "seckill-order-group", entries.getId());

            } catch (org.springframework.data.redis.RedisSystemException e) {
                log.warn("处理错误消息时Redis连接异常: {}", e.getMessage());
                retryCount++;
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    break;
                }
            } catch (Exception e) {
                log.error("处理错误消息失败", e);
                retryCount++;
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }

        if (retryCount >= MAX_RETRY) {
            log.warn("处理错误消息达到最大重试次数，放弃处理");
        }
    }

    //    @Override
//    public String seckill(Long voucherId) {
//        // 1. 获取用户ID
//        String userId = redisTemplate.opsForValue().get(LOGIN_USER_KEY);
//        log.info("当前用户是"+userId);
//        if (userId == null || userId.isEmpty()) {
//            return "未登录";
//        }
//
//        // 2. 用户级锁（确保一人一单）
//        RLock userLock = redissonClient.getLock(USER_ORDER_LOCK + userId + ":" + voucherId);
//        System.out.println("尝试获取用户锁"+String.valueOf(userLock));
//        log.info("尝试获取锁"+ userLock.getName());
//        try {
//            // 尝试立即获取锁，不等待
//            if (!userLock.tryLock(0, 15, TimeUnit.SECONDS)) {
//                return "您正在处理订单，请勿重复提交";
//            }
//
//            // 3. 商品级锁（防超卖）
//            RLock stockLock = redissonClient.getLock(STOCK_KEY + voucherId);
//            log.info("尝试获取商品锁" + stockLock.getName());
//            try {
//                if (!stockLock.tryLock(1, 10, TimeUnit.SECONDS)) {
//                    return "抢购人数过多，请稍后再试";
//                }
//
//                // 4. 执行原子操作
//                Long result = redisTemplate.execute(
//                        SECKILL_SCRIPT,
//                        Collections.emptyList(),
//                        userId,
//                        String.valueOf(voucherId)
//                );
//
//                if (result!=0) {
//                    return "系统错误";
//                }
//                // 5. 创建订单
//                return createOrder(Long.valueOf(userId), voucherId);
//            } finally {
//                if (stockLock.isHeldByCurrentThread()) {
//                    stockLock.unlock();
//                }
//            }
//        } catch (InterruptedException e) {
//            Thread.currentThread().interrupt();
//            return "系统繁忙";
//        } finally {
//            if (userLock.isHeldByCurrentThread()) {
//                userLock.unlock();
//            }
//        }
//    }

    /**
     * 抢购代金券
     *
     * @param voucherId
     * @return
     */
    @Override
    @Transactional
    public String seckill(Long voucherId) {
        //先判断用户是否登录
        String user = redisTemplate.opsForValue().get("login:user:");
        if (user == null || user.isEmpty()) {
            return "未登录";
        }
        //用户存在对商品用户加锁
        RLock lock = redissonClient.getLock("login:user:" + user + ":" + "voucherid" + voucherId);
        boolean locked=false;
        try {
             locked = lock.tryLock(1, 15, TimeUnit.SECONDS);
            if (!locked) {
                //锁未获取成功 添加等待时间
                redisTemplate.expire("login:user:" + user, 10, TimeUnit.MINUTES);
                return "您正在处理订单，请勿重复提交";
            }

//获取锁成功 对商品进行上锁并下单
//        RLock stocklock = redissonClient.getLock("seckill:stock:" + voucherId);
//        try {
//            if (!stocklock.tryLock(1, 10, TimeUnit.SECONDS)) {
//                //锁未获取成功 添加等待时间
//                redisTemplate.expire("seckill:stock:" + voucherId, 10, TimeUnit.MINUTES);
//                return "抢购人数过多，请稍后再试";
//            }
//        } catch (InterruptedException e) {
//            log.error("商品锁" + e.getMessage(), e);
//            throw new RuntimeException(e);
//        } finally {
//            stocklock.unlock();
//        }
            Long execute = redisTemplate.execute(SECKILL_SCRIPT,
                    Collections.emptyList(),
                    user,
                    String.valueOf(voucherId));
            if (execute != 0) {
                return execute == -1 ? "库存不足" : (execute == -2 ? "重复下单" : ("系统错误"));
            }
            asynccreateOrder(Long.valueOf(user), voucherId);
            return "抢购成功";
        } catch (InterruptedException e) {
            log.error("用户商品锁" + e.getMessage(), e);
            throw new RuntimeException(e);
        } finally {
            if (locked) {
                lock.unlock();
            }}

    }

    /**
     * 异步下单
     */
    @Transactional
    public void asynccreateOrder(Long userid, Long voucherId) {
        try {
            createOrder(userid, voucherId);
        } catch (Exception e) {
            log.error("异步下单失败");
            rollbackcreate(userid, voucherId);
        }

    }

    /**
     * 补偿机制
     *
     * @param userid
     * @param voucherId
     */
    private void rollbackcreate(Long userid, Long voucherId) {
        try {
            redisTemplate.opsForValue().increment("seckill:stock:" + voucherId);
            redisTemplate.opsForSet().remove("seckill:order", userid);
        } catch (Exception e) {
            log.error("回滚失败,voucherid={}", voucherId);
        }
    }

    /**
     * 生成订单
     *
     * @param userId
     * @param voucherId
     * @return
     */


    @Transactional
    public String createOrder(Long userId, Long voucherId) {
        //再次检查用户是否登录*
        String user = redisTemplate.opsForValue().get("login:user:");
        if (user == null || user.isEmpty()) {
            return "未登录";
        }
        //用户登录 创建订单
        //数据库兜底检查是否已经下单过
        boolean exsist = userVoucherService.lambdaQuery().eq(UserVoucher::getUserId, userId)
                .eq(UserVoucher::getVoucherId, voucherId)
                .count() > 0;
        if (exsist) {
            log.info("用户已经下单过");
            return "用户已经下单过";
        }
        HashMap<String, Object> msg=new HashMap<>();
        msg.put("userId",userId);
        msg.put("voucherId",voucherId);
        //发送消息异步创建订单
        rabbitTemplate.convertAndSend("order.topic", "订单创建成功",msg);

        return "下单成功";

    }


    //    @Transactional(
//            rollbackFor = Exception.class,
//            isolation = Isolation.SERIALIZABLE // 最高隔离级别
//    )
////    public String createOrder(Long userId, Long voucherId) {
////        // 1. 再次检查是否已下单（数据库兜底）
////        if (userVoucherService.lambdaQuery()
////                .eq(UserVoucher::getUserId, userId)
////                .eq(UserVoucher::getVoucherId, voucherId)
////                .count() > 0) {
////            // 补偿Redis库存
////            redisTemplate.opsForValue().increment(STOCK_KEY + voucherId);
////            return "您已经参与过本次抢购";
////        }
////
////        // 2. 创建订单
////        UserVoucher uv = new UserVoucher();
////        uv.setUserId(userId);
////        uv.setVoucherId(voucherId);
////        userVoucherService.save(uv);
////
////        // 3. 扣减数据库库存
////        boolean success = voucherService.lambdaUpdate()
////                .setSql("stock = stock - 1")
////                .eq(Voucher::getId, voucherId)
////                .gt(Voucher::getStock, 0)
////                .update();
////
////        if (!success) {
////            redisTemplate.opsForValue().increment(STOCK_KEY + voucherId);
////            throw new RuntimeException("库存不足");
////        }
////
////        return "抢购成功";
////    }
//
//    @Override

    /**
     * 添加代金券
     * @param voucher
     * @return
     */
    public ResponseEntity<Voucher> addSeckill(Voucher voucher) {
        voucherService.save(voucher);
        // 初始化库存
        redisTemplate.opsForValue().set(
                STOCK_KEY + voucher.getId(),
                String.valueOf(voucher.getStock()),
                24, TimeUnit.HOURS
        );
        // 初始化订单集合
        redisTemplate.opsForSet().add(
                ORDER_KEY + voucher.getId(),
                "INIT_VALUE"
        );
        redisTemplate.expire(
                ORDER_KEY + voucher.getId(),
                24, TimeUnit.HOURS
        );
        return ResponseEntity.ok(voucher);
    }

    /**
     * 消息队列处理订单信息
     * @param userVoucher
     */
    @Override
    public void handlerOrderInfo(UserVoucher userVoucher) {
        Long userid =userVoucher.getUserId();
        Long voucherId = userVoucher.getVoucherId();
        RLock lock = redissonClient.getLock("msg:" + userid + ":" + voucherId);
        boolean locked = false;
        try {
             locked = lock.tryLock(1, 5, TimeUnit.SECONDS);
            if (!locked) {
                log.info("获取消息失败");
                return;
            }
            try {
                createOrder(Long.valueOf(userid), voucherId);
            } finally {
                if (locked) {
                    lock.unlock();
                }
            }
        } catch (InterruptedException e) {
            log.info("获取消息失败");
            Thread.currentThread().interrupt();
            //补偿
            rollbackcreate(userid, voucherId);
        }
    }
}